package com.yinanxi.cool.dws

import TestTree.folowPv2
import com.yinanxi.cool.SparkUtils
import org.apache.spark.rdd.RDD

import scala.collection.mutable.ListBuffer

/**
 * @Created by Orion
 * @Description
 */
case class PageView(pageUrl: String, guid: Long, sessionId: String, accLong: Long, refUrl: String, ts: Long)

case class PageNode(pg: PageView, children: ListBuffer[PageNode])

object PageFollowPvJob {
  def main(args: Array[String]): Unit = {

    System.setProperty("HADOOP_USER_NAME" , "root")
    val session = SparkUtils.getSesion()
    // 加载数据   页面明细数据
    val df = session.read.table("dws.cool_app_page_dtl").where("dt=' 2023-02-16'")
    // 将表数据构建成一个RDD  对RDD操作灵活
    val rdd1 = df.rdd.map(row => { // 解析行数
      val pageUrl = row.getAs[String]("page_url")
      val guid = row.getAs[Long]("guid")
      val sessionId = row.getAs[String]("sess_id")
      val accLong = row.getAs[Long]("acc_long")
      val refUrl = row.getAs[String]("ref_url")
      val ts = row.getAs[Long]("ts")
      PageView(pageUrl, guid, sessionId, accLong, refUrl, ts)
    })
    // 以session为单位 构建树
    val grouped: RDD[(String, Iterable[PageView])] = rdd1.groupBy(_.sessionId)

    // 处理每个会话
    val roots: RDD[PageNode] = grouped.flatMap(tp => {
      val sessionId = tp._1
      // 将一次会话请求的页面进行时间先后排序
      val sorted = tp._2.toList.sortBy(_.ts)
      var trees = ListBuffer[ListBuffer[PageNode]]()
      var tree: ListBuffer[PageNode] = null
      // 遍历一次会话请求的所有页面
      for (pageView <- sorted) {
        // 每条数据就是一个节点
        val currentNode = PageNode(pageView, ListBuffer[PageNode]())
        // 1) tree == null   2) 没有来源   3)  外部来源
        if (tree == null || pageView.refUrl == null || pageView.refUrl.trim.equals("") || pageView.refUrl.startsWith("http")) { // 是否需求构建树
          tree = ListBuffer[PageNode]()
          // 将当前节点挂载到树上    根节点
          tree.append(currentNode)
          // 将 树加载到会话下的树林里
          trees += tree
        } else { // 不用构建树
          tree += currentNode
          // 给当前节点找到正确的父节点
          if (pageView.refUrl != null && pageView.refUrl.startsWith("/")) { //  内部跳转
            //  给当前节点找到正确的父节点
            val maybeNode = tree.reverse.find(node => node.pg.pageUrl.equals(pageView.refUrl))
            if (maybeNode.isDefined) {
              // 挂载
              maybeNode.get.children += currentNode
            } else {
              // 将当前节点挂载到最后
              if (tree.size > 1) { //tree中至少有两个节点
                tree(tree.size - 2).children += currentNode
              }
            }
          }
        }
      }
      // 将结果返回  返回树   返回树的根节点
      trees.map(tree => tree.head) //
    })
    // 统计每个节点的子节点个数   下游页面贡献量
    val res: RDD[(String, String, Long, String, Long, Long, Int)] = roots.flatMap(root => {
      val buffer = ListBuffer[(PageView, Int)]()
      // 调用方法
      folowPv(root, buffer)
      buffer.map(tp => {
        val pageView = tp._1
        val pv = tp._2
        val refUrl = pageView.refUrl
       // 处理  refUrl   确定来源类型
        (pageView.pageUrl, pageView.refUrl, pageView.guid, pageView.sessionId, pageView.accLong, pageView.ts, pv)

      })
    })
    // 测试  代码逻辑没问题
    res.take(10).foreach(println)
    // 将计算的逻辑编写在代码中      tmp  join   cool_app_acc_user_dtl
    import  session.implicits._
    val pageDf = res.toDF("page_url", "ref_url", "guid", "sess_id", "acc_long", "ts", "follow_pv")
   // 使用代码分析 ref_url   ref_type赋值
    pageDf.createOrReplaceTempView("page_dtl")

    session.sql(
      """
        | -- 宽表数据
        |insert  into  table  dws.cool_app_acc_page_wide partition(dt=' 2023-02-16')
        |select
        |t1.page_url  ,
        |t1.guid  ,
        |t1.sess_id ,
        |t1.ref_url ,
        |case
        | when t1.ref_url is null or trim(t1.ref_url) == '' then  '直接访问'
        | when t1.ref_url like '/%'  then '内部跳转'
        | when parse_url(t1.ref_url , 'HOST') in ('www.baidu.com' , 'www.gougou.com' , 'www.sougou.com' , 'www.360.com' , 'www.google.com') then parse_url(t1.ref_url , 'HOST')
        | else  '其他'
        | end  as  ref_type ,
        | if(t2.guid is not null , t2.is_new , 0) as  is_new ,
        | if(row_number() over(partition by  sess_id order by ts) > 1 , 0 ,1) as  is_enter  ,
        | if(row_number() over(partition by  sess_id order by ts desc) > 1 , 0 ,1) as  is_exit  ,
        | t1.follow_pv ,
        | t1.acc_long ,
        | t1.ts
        |from
        |page_dtl t1
        |left join
        |(
        | select
        | guid ,
        | is_new
        | from
        | dws.cool_app_acc_user_dtl where dt = ' 2023-02-16'
        |) t2
        |on  t1.guid = t2.guid
        |""".stripMargin)
    /*.write.partitionBy("dt").format("orc")/*option("orc.compress", "snappy")*/.saveAsTable("dws.cool_app_acc_page_wide")
*/

  }

  /**
   * 页面下游贡献量
   *
   * @param node
   * @param buffer
   * @return
   */
  def folowPv(node: PageNode, buffer: ListBuffer[(PageView, Int)]): Int = {
    var pv = 0
    val children = node.children
    if (children != null && children.size > 0) {
      pv += children.size
      // 遍历所有的子节点
      for (elem <- children) {
        pv += folowPv(elem, buffer)
      }
    }
    // 将当前节点的地址和子节点个数存储在缓存中
    buffer.append((node.pg, pv))
    pv
  }


}
